package com.aiwiown.snackmq.broker.handler.console;

import com.aiwiown.snackmq.broker.auth.AuthorizationService;
import com.aiwiown.snackmq.common.auth.Action;
import com.aiwiown.snackmq.common.handler.MessageHandler;
import com.aiwiown.snackmq.common.handler.RequestContext;
import com.aiwiown.snackmq.common.message.Message;
import com.aiwiown.snackmq.common.message.MessageStatus;
import com.aiwiown.snackmq.common.message.MessageType;
import com.aiwiown.snackmq.common.protocol.TopicPartition;
import com.aiwiown.snackmq.common.serialization.JsonSerializer;
import com.aiwiown.snackmq.common.serialization.SerializerFactory;
import com.aiwiown.snackmq.storage.StorageService;
import com.aiwiown.snackmq.storage.exception.PartitionNotFoundException;
import com.aiwiown.snackmq.storage.partition.LogPartition;
import com.aiwiown.snackmq.storage.service.PartitionedStorageService;
import lombok.extern.slf4j.Slf4j;

import java.io.File;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.NavigableMap;

/**
 * 处理分区诊断请求的处理器。
 * 该处理器负责分析指定分区的段文件和索引信息，帮助诊断数据一致性问题。
 */
@Slf4j
public class DiagnosePartitionHandler implements MessageHandler {

    private final StorageService storageService;
    private final AuthorizationService authorizationService;
    private final JsonSerializer serializer = SerializerFactory.getJsonSerializer();

    public DiagnosePartitionHandler(StorageService storageService, AuthorizationService authorizationService) {
        this.storageService = storageService;
        this.authorizationService = authorizationService;
    }

    @Override
    public void handle(RequestContext ctx, Message request) {
        try {
            String topic = request.getTopic();
            int partitionId = Integer.parseInt(request.getProperty("partition"));

            // 授权检查
            authorizationService.authorize(
                    request.getProperty("authToken"),
                    Action.CONSUME,
                    new AuthorizationService.Resource("TOPIC", topic)
            );

            Map<String, Object> diagnosis = diagnosePartition(topic, partitionId);

            Message response = Message.builder()
                    .messageId(request.getMessageId())
                    .type(MessageType.RESPONSE)
                    .status(MessageStatus.SUCCESS)
                    .body(serializer.serialize(diagnosis))
                    .build();
            ctx.writeResponse(response);

        } catch (PartitionNotFoundException e) {
            log.warn("Partition not found for diagnosis: {}", request, e);
            Message errorResponse = Message.builder()
                    .messageId(request.getMessageId())
                    .type(MessageType.RESPONSE)
                    .status(MessageStatus.FAILED)
                    .errorMessage("Partition not found: " + e.getMessage())
                    .build();
            ctx.writeResponse(errorResponse);
        } catch (Exception e) {
            log.error("Failed to diagnose partition for request: {}", request, e);
            Message errorResponse = Message.builder()
                    .messageId(request.getMessageId())
                    .type(MessageType.RESPONSE)
                    .status(MessageStatus.FAILED)
                    .errorMessage("Diagnosis failed: " + e.getMessage())
                    .build();
            ctx.writeResponse(errorResponse);
        }
    }

    private Map<String, Object> diagnosePartition(String topic, int partitionId) throws Exception {
        Map<String, Object> diagnosis = new HashMap<>();
        
        if (!(storageService instanceof PartitionedStorageService)) {
            throw new UnsupportedOperationException("Diagnosis only supported for PartitionedStorageService");
        }

        PartitionedStorageService partitionedStorage = (PartitionedStorageService) storageService;
        TopicPartition topicPartition = new TopicPartition(topic, partitionId);
        
        // 获取分区信息
        LogPartition logPartition = partitionedStorage.getLogPartition(topicPartition);
        if (logPartition == null) {
            throw new PartitionNotFoundException("Partition " + topicPartition + " is not active on this broker.");
        }

        // 基本信息
        diagnosis.put("topic", topic);
        diagnosis.put("partitionId", partitionId);
        diagnosis.put("partitionPath", logPartition.getPartitionPath());
        diagnosis.put("segmentSize", logPartition.getSegmentSize());
        diagnosis.put("indexInterval", logPartition.getIndexInterval());
        diagnosis.put("nextOffset", logPartition.getNextOffset());

        // 检查损坏的段
        List<Long> corruptedSegments = partitionedStorage.getCorruptedSegments(topicPartition);
        diagnosis.put("corruptedSegments", corruptedSegments);
        diagnosis.put("hasCorruptedSegments", !corruptedSegments.isEmpty());

        // 段信息
        NavigableMap<Long, com.aiwiown.snackmq.storage.segment.LogSegment> segments = logPartition.getSegments();
        Map<String, Object> segmentsInfo = new HashMap<>();
        for (Map.Entry<Long, com.aiwiown.snackmq.storage.segment.LogSegment> entry : segments.entrySet()) {
            Long baseOffset = entry.getKey();
            com.aiwiown.snackmq.storage.segment.LogSegment segment = entry.getValue();
            
            Map<String, Object> segmentInfo = new HashMap<>();
            segmentInfo.put("baseOffset", baseOffset);
            segmentInfo.put("entryCount", segment.getEntryCount().get());
            segmentInfo.put("logPosition", segment.getLogPosition().get());
            segmentInfo.put("isActive", segment == logPartition.getActiveSegment());
            segmentInfo.put("isCorrupted", segment.isIndexCorrupted());
            
            // 文件信息
            File logFile = new File(segment.getLogFile().getParent(), String.format("%020d.log", baseOffset));
            File indexFile = new File(segment.getLogFile().getParent(), String.format("%020d.index", baseOffset));
            segmentInfo.put("logFileExists", logFile.exists());
            segmentInfo.put("logFileSize", logFile.exists() ? logFile.length() : 0);
            segmentInfo.put("indexFileExists", indexFile.exists());
            segmentInfo.put("indexFileSize", indexFile.exists() ? indexFile.length() : 0);
            
            segmentsInfo.put("segment_" + baseOffset, segmentInfo);
        }
        diagnosis.put("segments", segmentsInfo);

        // 测试读取一些关键offset
        Map<String, Object> testReads = new HashMap<>();
        for (Long baseOffset : segments.keySet()) {
            // 测试段开始
            Message startMessage = logPartition.read(baseOffset);
            testReads.put("offset_" + baseOffset + "_start", startMessage != null ? "SUCCESS" : "FAILED");
            
            // 测试段结束（如果存在）
            com.aiwiown.snackmq.storage.segment.LogSegment segment = segments.get(baseOffset);
            long endOffset = baseOffset + segment.getEntryCount().get() - 1;
            if (endOffset >= baseOffset) {
                Message endMessage = logPartition.read(endOffset);
                testReads.put("offset_" + endOffset + "_end", endMessage != null ? "SUCCESS" : "FAILED");
            }
        }
        diagnosis.put("testReads", testReads);

        return diagnosis;
    }

    @Override
    public MessageType getMessageType() {
        return MessageType.DIAGNOSE_PARTITION_REQUEST;
    }
} 